package com.xrui.bigdataaccessor;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;

public class HDFSAccessor {

    public static void hdfsReader(boolean isKerberosEnable, String kerberosLoginUser,
                                  String hdfsUri,
                                  String hdfsFile,
                                  InputStream inputStream )
    {
        Configuration config = new Configuration();
        FileSystem hdfs = null;
        try {
            if (isKerberosEnable){
                System.setProperty("java.security.krb5.conf", "./config/kerberos/krb5.conf");
                config.set("hadoop.security.authentication","kerberos");
                config.addResource("./config/cluster/hadoop/core-site.xml");
                config.addResource("./config/cluster/hadoop/hdfs-site.xml");
                UserGroupInformation.setConfiguration(config);
                UserGroupInformation.loginUserFromKeytab(kerberosLoginUser, "./config/kerberos/hdfs.keytab");
                UserGroupInformation userGroupInformation = UserGroupInformation.getLoginUser();
            }
            hdfs = FileSystem.get(URI.create(hdfsUri), config);
            FSDataOutputStream outputStream = hdfs.create(new Path(hdfsFile), true);
            IOUtils.copyBytes(inputStream, outputStream, 4096, true);
        }catch(IOException ioe){

        }

    }


}
